package ds_industry_2025.ds.ds01.T4

import org.apache.spark.ml.{Pipeline, feature}
import org.apache.spark.ml.feature.{OneHotEncoder, StandardScaler, VectorAssembler}
import org.apache.spark.ml.linalg.{DenseVector, SparseVector}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

import java.util.Properties
/*
      2、根据Hive的dwd库中相关表或MySQL中shtd_store中相关商品表（sku_info），获取id、spu_id、price、weight、tm_id、
      category3_id 这六个字段并进行数据预处理，对price、weight进行规范化(StandardScaler)处理，对spu_id、tm_id、
      category3_id进行one-hot编码处理（若该商品属于该品牌则置为1，否则置为0）,并按照id进行升序排序，在集群中输出第一条数据前
      10列（无需展示字段名），将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
 */
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")


    //  todo 定义三个字段的排序规则
    val window1 = Window.orderBy("spu_id")
    val window2 = Window.orderBy("tm_id")
    val window3 = Window.orderBy("category3_id")

    // todo 将spu_id,tm_id,category3_id三列分类的特征转化为数值型(索引型)的，为了方便下面的机器学习或者one-hot编码做数据准备
    val source = spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "sku_info", conn)
      .select("id", "spu_id", "price", "weight", "tm_id", "category3_id")
      .withColumn("spu_id_index", dense_rank().over(window1) - 1)
      .withColumn("tm_id_index", dense_rank().over(window2) - 1)
      .withColumn("category3_id_index", dense_rank().over(window3) - 1)

//    source.show(false)
    //  todo 将这张表保存，给下一题的推荐系统使用
    source.write.format("hive").mode("overwrite")
      .saveAsTable("tzgc.t1")

    //  todo OneHotEncoder，用于将分类特征的索引列转换为 One-Hot 编码。
    //  todo one-hot编码处理后的每个字段都是返回一个稀疏向量
    //  todo setDropLast(false)最好添加，可以便面出问题
    val hotEncoder = new OneHotEncoder()
      .setInputCols(Array("spu_id_index","tm_id_index","category3_id_index"))
      .setOutputCols(Array("spu_id_hot","tm_id_hot","category3_id_hot"))
      .setDropLast(false)

    // todo VectorAssembler的设计目的是将多个列组合成一个向量列,所以输入列是Array这样的数组
    //  在使用StandardScaler规范化之前需要把数据转化为特征向量
    val vectorAssembler1 = new VectorAssembler()
      .setInputCols(Array("price"))
      .setOutputCol("price_v")

    val vectorAssembler2=new VectorAssembler()
      .setInputCols(Array("weight"))
      .setOutputCol("weight_v")

    //  todo 规范化器StandardScaler
    //  setWithMean(true)：在标准化过程中减去均值。这意味着数据会被中心化，均值为 0。
    val standardScaler1 = new StandardScaler()
      .setInputCol("price_v")
      .setOutputCol("price_sca")
      .setWithMean(true)

    val standardScaler2=new StandardScaler()
      .setInputCol("weight_v")
      .setOutputCol("weight_sca")
      .setWithMean(true)

    //  todo 定义流水线，按照顺序执行操作,并且训练模型
    val piplelineModel = new Pipeline()
      .setStages(Array(vectorAssembler1,vectorAssembler2,standardScaler1,standardScaler2,hotEncoder))
      .fit(source)

    //  todo 注册两个函数分别处理稀疏向量和密集向量()
    //  稀疏向量:将稀疏向量转化为数组，然后以逗号分割的字符串
    spark.udf.register(
      "vectorToArray",
      (v1:SparseVector) => { v1.toArray.mkString(",") }
    )
    //  密集向量:拿到密集向量的第一个数值,因为标准化处理就是得到一个数组，里面是一个数据
    spark.udf.register(
      "vectorToDouble",
      (v1:DenseVector) => v1.apply(0)
    )


    val result = piplelineModel.transform(source)
      .withColumn("spu_id_hot", expr("vectorToArray(spu_id_hot)"))
      .withColumn("tm_id_hot", expr("vectorToArray(tm_id_hot)"))
      .withColumn("category3_id_hot", expr("vectorToArray(category3_id_hot)"))
      .withColumn("price_sca", expr("vectorToDouble(price_sca)"))
      .withColumn("weight_sca", expr("vectorToDouble(weight_sca)"))
      .select("id", "price_sca", "weight_sca","spu_id_hot", "tm_id_hot", "category3_id_hot")
      .orderBy(asc("id"))
      .limit(1)


    println("---------------------第一条数据前10列的结果展示为------------------")
    result.collect().foreach(
      r => {
        println(r.toSeq.flatMap( r => r.toString.split(",")).take(10).mkString(","))
      }
    )




    spark.close()

  }

}
